package com.mini.rpc.provider.registry;

import com.mini.rpc.common.RpcServiceHelper;
import com.mini.rpc.common.ServiceMeta;
import com.mini.rpc.provider.registry.loadbalancer.LoadBalancerFactory;
import com.mini.rpc.provider.registry.loadbalancer.LoadBalancerType;
import com.mini.rpc.provider.registry.loadbalancer.ServiceLoadBalancer;
import com.mini.rpc.provider.registry.loadbalancer.ZKConsistentHashLoadBalancer;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;

import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
public class ZookeeperRegistryService implements RegistryService {
    public static final int BASE_SLEEP_TIME_MS = 1000;
    /**
     * 最大重试策略
     */
    public static final int MAX_RETRIES = 3;
    public static final String ZK_BASE_PATH = "/mini_rpc";

    private final ServiceDiscovery<ServiceMeta> serviceDiscovery;

    /**
     * 构造器里面构建 Zookeeper 的客户端
     */
    public ZookeeperRegistryService(String registryAddr) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.newClient(registryAddr,
                new ExponentialBackoffRetry(BASE_SLEEP_TIME_MS, MAX_RETRIES));
        client.start();
        JsonInstanceSerializer<ServiceMeta> serializer = new JsonInstanceSerializer<>(ServiceMeta.class);
        this.serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceMeta.class)
                .client(client)
                .serializer(serializer)
                .basePath(ZK_BASE_PATH)
                .build();
        this.serviceDiscovery.start();
    }

    @Override
    public void register(ServiceMeta serviceMeta) throws Exception {
        // ServiceInstance代表了一个服务实例
        ServiceInstance<ServiceMeta> serviceInstance = ServiceInstance
                .<ServiceMeta>builder()
                // 唯一标识Id
                .name(RpcServiceHelper.buildServiceKey(serviceMeta.getServiceName(), serviceMeta.getServiceVersion()))
                .address(serviceMeta.getServiceAddr())
                .port(serviceMeta.getServicePort())
                .payload(serviceMeta)
                .build();
        serviceDiscovery.registerService(serviceInstance);
    }

    @Override
    public void unRegister(ServiceMeta serviceMeta) throws Exception {
        ServiceInstance<ServiceMeta> serviceInstance = ServiceInstance
                .<ServiceMeta>builder()
                .name(serviceMeta.getServiceName())
                .address(serviceMeta.getServiceAddr())
                .port(serviceMeta.getServicePort())
                .payload(serviceMeta)
                .build();
        serviceDiscovery.unregisterService(serviceInstance);
    }

    /**
     * 服务发现
     *
     * @param serviceName     服务名称, 例如: com.mini.rpc.provider.facade.HelloFacade#1.0.0
     * @param invokerHashCode 客户端的hashCode
     * @return 服务节点元数据
     */
    @Override
    public ServiceMeta discovery(String serviceName, int invokerHashCode) throws Exception {
        Collection<ServiceInstance<ServiceMeta>> serviceInstances = serviceDiscovery.queryForInstances(serviceName);
        // todo 这里的负载均衡算法写死了, 配置文件写类全路径名, 通过反射创建
        //      思考, 这个负载均衡算法好像是针对于Zookeeper注册中心来的
        List<ServiceMeta> serviceMetaList = serviceInstances.stream()
                .map(ServiceInstance::getPayload).collect(Collectors.toList());

        return new ZKConsistentHashLoadBalancer().select(serviceMetaList, invokerHashCode);
    }

    /**
     * 服务发现
     *
     * @param serviceName  服务名称, 例如: com.rpc.service.HelloImpl#1.0.0
     * @param loadBalancer 负载均衡算法
     * @return 服务节点元数据
     */
    @Override
    public ServiceMeta discovery(String serviceName, String loadBalancer, int invokerHashCode) throws Exception {
        Collection<ServiceInstance<ServiceMeta>> serviceInstances = serviceDiscovery.queryForInstances(serviceName);

        List<ServiceMeta> serviceMetaList = serviceInstances.stream()
                .map(ServiceInstance::getPayload).collect(Collectors.toList());
        // serviceMetaList
        ServiceLoadBalancer<ServiceMeta> balancer = LoadBalancerFactory
                .getLoadBalancer(LoadBalancerType.valueOf(loadBalancer));
        return balancer.select(serviceMetaList, invokerHashCode);
    }

    @Override
    public void destroy() throws IOException {
        log.info(">>> {} registry destroy", getRegistryName());
        serviceDiscovery.close();
    }
}
